"""
Case Type   : 发布订阅
Case Name   : 发布订阅过程中删除发布订阅
Create At   : 2022-11-14
Owner       : peilinqian
Description :
    1.两个集群创建表
    2.创建发布订阅
    3.发布端更新数据,更新过程中删除发布端
    4.订阅端查询数据同步
    5.等待集群A数据插入成功后,重建发布端
    6.发布端更新数据,订阅端查询数据同步
    7.发布端更新数据,更新过程中删除订阅端
    8.订阅端查询数据同步
    9.重建订阅端
    10.发布端插入数据,订阅端检查数据同步状态
    11.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.数据未同步
    5.成功
    6.成功,数据同步
    7.成功
    8.数据未同步,日志报错订阅不存在
    9.成功
    10.发布端插入数据成功,订阅端step6数据不存在,step10数据同步
    11.成功
History     : 
"""

import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Common import Common
from testcase.utils.ComThread import ComThread
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote_PrimaryDbUser')
        self.com_pub = Common()
        self.tb_name = 't_pub_sub_reliability_0008'
        self.subname = "s_pub_sub_reliability_0008"
        self.pubname = "p_pub_sub_reliability_0008"
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE)
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        result = self.commsh_pub.executDbSql("show wal_level;")
        self.wal_level = str(result.splitlines()[-2]).strip()
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE, 'pg_hba.conf')}" \
            f" {os.path.join(self.parent_path_sub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        self.hostname = self.pri_userdb_sub.sh('hostname').result().strip()
        text = "--step:预置条件,修改pg_hba expect:成功"
        self.log.info(text)
        guc_res = self.commsh_pub.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_pub.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    all  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        result = self.commsh_pub.executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
        self.assertTrue(result, '执行失败' + text)
        pub_result = self.commsh_pub.stopDbCluster(get_detail=True)
        pub_result1 = self.commsh_pub.startDbCluster(get_detail=True)
        self.assertTrue(pub_result, '执行失败' + text)
        flg = self.constant.START_SUCCESS_MSG in pub_result1 or 'Degrade' in \
              pub_result1
        self.assertTrue(flg, '执行失败' + text)
        guc_res = self.commsh_sub.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_sub.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE,
            f'host all  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)

    def test_pub_sub(self):
        text = "--step1:在两个集群创建表 expect:成功--"
        self.log.info(text)
        sql_cmd = f"drop table if exists {self.tb_name};" \
            f"create table {self.tb_name}(c_int int primary key,c_text text);"
        pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败' + text)
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertEqual(sub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败' + text)

        text = "--step2:创建发布订阅 expect:成功--"
        self.log.info(text)
        sql_cmd = f"drop publication if exists {self.pubname};" \
            f"create publication {self.pubname} for all tables;"
        pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.create_pub_succ_msg, pub_result,
                      '执行失败' + text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE)
        self.assertIn(self.constant.create_keycipher_success, result,
                      '执行失败' + text)
        sql_cmd = f"drop subscription if exists {self.subname};" \
            f"create subscription {self.subname} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' publication " \
            f"{self.pubname};"
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertIn(self.constant.create_sub_succ_msg, sub_result, '执行失败'
                      + text)

        text = "--step3:发布端更新数据,更新过程中删除发布端 expect:成功--"
        self.log.info(text)
        sql_cmd = f"insert into {self.tb_name}(c_int) values" \
            f"(generate_series(1,1000000));"
        pub_thread1 = ComThread(self.commsh_pub.executDbSql, args=(
            sql_cmd, self.user_param_pub,))
        pub_thread1.setDaemon(True)

        sql_cmd = f"drop publication {self.pubname};"
        pub_thread2 = ComThread(self.commsh_pub.executDbSql, args=(
            sql_cmd, self.user_param_pub,))
        pub_thread2.setDaemon(True)
        pub_thread1.start()
        pub_thread2.start()
        pub_thread1.join(30)
        pub_thread2.join(10)
        pub_result1 = pub_thread1.get_result()
        self.log.info(pub_result1)
        pub_result2 = pub_thread2.get_result()
        self.log.info(pub_result2)
        self.assertIn(self.constant.insert_success_msg, pub_result1, '执行失败'
                      + text)
        self.assertIn(self.constant.drop_pub_succ_msg, pub_result2, '执行失败'
                      + text)

        text = "--step4:订阅端查询数据同步 expect:数据未同步--"
        self.log.info(text)
        sql_cmd = f"select pg_sleep(5.5);select count(*) from {self.tb_name};"
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertEqual(str(sub_result.splitlines()[-2]).strip(), '0',
                         '执行失败' + text)

        text = "--step5:等待集群A数据插入成功后,重建发布端  expect:成功--"
        self.log.info(text)
        sql_cmd = f"create publication {self.pubname} for all tables;"
        pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.create_pub_succ_msg, pub_result,
                      '执行失败' + text)

        text = "--step6:发布端更新数据,订阅端查询数据同步 expect:成功," \
               "数据同步--"
        self.log.info(text)
        sql_cmd = f"truncate table {self.tb_name};"
        pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertIn('TRUNCATE TABLE', sub_result, '执行失败' + text)
        self.assertIn('TRUNCATE TABLE', pub_result, '执行失败' + text)
        sql_cmd = f"insert into {self.tb_name}(c_int) values" \
            f"(generate_series(1,1000000));"
        pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.insert_success_msg, pub_result, ' 执行失败'
                      + text)
        sql_cmd = f"select pg_sleep(5.5);select count(*) from {self.tb_name};"
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertEqual(str(sub_result.splitlines()[-2]).strip(), '2000000',
                         '执行失败' + text)

        text = "--step7:发布端更新数据,更新过程中删除订阅端 expect:成功--"
        self.log.info(text)
        sql_cmd = f"delete from {self.tb_name} where c_int < 100000;"
        pub_thread3 = ComThread(self.commsh_pub.executDbSql, args=(
            sql_cmd, self.user_param_pub,))
        pub_thread3.setDaemon(True)

        sql_cmd = f"drop subscription {self.subname};"
        sub_thread = ComThread(self.commsh_sub.executDbSql, args=(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE,))
        sub_thread.setDaemon(True)
        pub_thread3.start()
        sub_thread.start()
        pub_thread3.join(30)
        sub_thread.join(15)
        pub_result3 = pub_thread3.get_result()
        self.log.info(pub_result3)
        sub_result = sub_thread.get_result()
        self.log.info(sub_result)
        self.assertIn(self.constant.delete_success_msg, pub_result3, '执行失败'
                      + text)
        self.assertIn(self.constant.drop_sub_succ_msg, sub_result, '执行失败'
                      + text)

        text = "--step8:订阅端查询数据同步 expect:数据未同步," \
               "日志报错订阅不存在--"
        self.log.info(text)
        sql_cmd = f"select pg_sleep(5.5);select count(*) from {self.tb_name} " \
            f"where c_int < 100000;"
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertEqual(str(sub_result.splitlines()[-2]).strip(), '0',
                         '执行失败' + text)

        text = "--step9:重建订阅端  expect:成功--"
        self.log.info(text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE)
        self.assertIn('', result, '执行失败:' + text)
        sql_cmd = f"create subscription {self.subname} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' publication " \
            f"{self.pubname};"
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertIn(self.constant.create_sub_succ_msg, sub_result, '执行失败'
                      + text)

        text = "--step10:发布端插入数据,订阅端检查数据同步状态 " \
               "expect:发布端插入数据成功,订阅端step6数据不存在," \
               "step10数据同步--"
        self.log.info(text)
        sql_cmd = f"select count(*) from {self.tb_name};delete from " \
            f"{self.tb_name} where c_int < 100000;"
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertEqual(str(sub_result.splitlines()[-2]).strip(), '100000',
                         '执行失败' + text)
        self.assertIn('', pub_result, '执行失败' + text)
        sql_cmd = f"delete from {self.tb_name};"
        pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.delete_success_msg, pub_result, '执行失败'
                      + text)
        sql_cmd = f"select pg_sleep(5.5);select count(*) from {self.tb_name};"
        sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(sub_result)
        self.assertEqual(str(sub_result.splitlines()[-2]).strip(), '0',
                         '执行失败' + text)

    def tearDown(self):
        text = "--step11:清理环境 expect:成功--"
        self.log.info(text)
        sql_cmd = f"drop subscription if exists {self.subname};"
        drop_sub_result = self.commsh_sub.executDbSql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(drop_sub_result)
        sql_cmd = f"drop publication if exists {self.pubname};"
        drop_pub_result = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)
        sql_cmd = f"drop table if exists {self.tb_name};"
        result_sub = \
            self.commsh_sub.executDbSql(
                sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE)
        self.log.info(result_sub)
        result_pub = self.commsh_pub.executDbSql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(result_pub)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')} "\
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} "
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_sub, 'pg_hba.conf')} "\
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE, 'pg_hba.conf')} "
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_pub.executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(result)
        pub_result = self.commsh_pub.stopDbCluster(get_detail=True)
        pub_result1 = self.commsh_pub.startDbCluster(get_detail=True)
        self.assertTrue(pub_result, '执行失败' + text)
        self.assertTrue(pub_result1, '执行失败' + text)
        self.assertTrue(result, '执行失败' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_pub, '执行失败'
                      + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_sub, '执行失败'
                      + text)
        self.log.info(f"-----{os.path.basename(__file__)} end-----")

